
from typing import Dict
from common import convert_to_timestamp, market_to_str
from converter import subscribe_rt_data_to_topic
from x_com import topic, xport_core
from x_com import source_api
import x_com_lib
import rtdata
import asyncio

logger = xport_core.getLogger()


class RealtimeKdataService:
    playing_data_dict: Dict[str, bool] = {}

    async def on_init(self):
        self.data_path = x_com_lib.config.get_str("data-path")
        print(self.data_path)
        logger.info("on_init...")
        print("初始化完成...")

    #
    async def on_finalize(self):
        logger.info("on_finalize...")

    # SubscribeRtData
    async def subscribe_rt_data(self, ctx: xport_core.Context,  param: source_api.SubscribeRTDataRequest):
        logger.info(f"param = {param.code}")
        subscibe_info = subscribe_rt_data_to_topic(ctx.conn_id, param)
        

        x_com_lib.stream.xport_subscribe_topic(subscibe_info)

    # CancelSubscribeRtData
    async def cancel_subscribe_rt_data(self, ctx: xport_core.Context,  param: source_api.SubscribeRTDataRequest):
        subscibe_info = subscribe_rt_data_to_topic(ctx.conn_id, param)
        x_com_lib.stream.xport_cancel_subscribe_topic(subscibe_info)

    async def play_data(self, market, code, date):
        # 20241105
        year = int(date / 10000)
        # 202411
        year_month = int(date / 100)
        market_str = market_to_str(market)
        lower_market_str = market_str.lower()
        tick_datas = rtdata.get_tick(f"{self.data_path}/tick/{year}/{year_month}/{date}/{lower_market_str}/tick_{lower_market_str}_{code}_{date}.z")

        tick_data_len = len(tick_datas)

        #
        rt_tick_data = source_api.RTTickData()
        rt_tick_data.market = market
        rt_tick_data.code = code
        rt_tick_datas = source_api.RTTickDatas()
        rt_tick_datas.datas.append(rt_tick_data)

        
        tag = f"{market_str}.{code}"
        logger.info(f"push tag = {tag}")
        for i in range(0, tick_data_len):
            # 跳过
            tick_data = tick_datas[i]
            # 赋值数据
            rt_tick_data.datetime = tick_data.datetime
            rt_tick_data.prev_close = tick_data.prev_close
            rt_tick_data.num_trades = tick_data.num_trades
            rt_tick_data.volume = tick_data.volume
            rt_tick_data.total_turnover = tick_data.total_turnover
            rt_tick_data.trading_phase_code = tick_data.trading_phase_code
            rt_tick_data.last = tick_data.last
            rt_tick_data.open = tick_data.open
            rt_tick_data.high = tick_data.high
            rt_tick_data.low = tick_data.low
            rt_tick_data.close = tick_data.close
            rt_tick_data.open_interest = tick_data.open_interest
            rt_tick_data.prev_settlement = tick_data.prev_settlement
            rt_tick_data.limit_up = tick_data.limit_up
            rt_tick_data.limit_down = tick_data.limit_down
            rt_tick_data.ask = tick_data.ask
            rt_tick_data.bid = tick_data.bid
            rt_tick_data.ask_vol = tick_data.ask_vol
            rt_tick_data.bid_vol = tick_data.bid_vol
            rt_tick_data.settlement = tick_data.settlement
            topic.rt_tick_data_with_tag(rt_tick_datas, tag)
            if i < tick_data_len - 1:
                next_tick = tick_datas[i + 1]
                #  因为 datetime 是到毫秒的，而sleep 是到秒的，因此要 除以 1000
                next_datetime = convert_to_timestamp(f"{next_tick.datetime}")
                cur_datetime = convert_to_timestamp(f"{tick_data.datetime}")
                wait_time = next_datetime - cur_datetime
                if wait_time > 0.5:
                  wait_time = 0.5
                await asyncio.sleep(wait_time)  # 模拟处理延时

        logger.info(f"推送行情{tag} ，长度：{tick_data_len} 完毕...")
        self.playing_data_dict.pop(tag, None)

    # PlayTickData
    async def play_tick_data(self, ctx: xport_core.Context,  param: source_api.RTDataRequest):
        market_str = market_to_str(param.market)
        key = f"{market_str}.{param.code}"

        is_playing = self.playing_data_dict.get(key)
        logger.info(f"is_playing = {is_playing}, key = {key}")
        if is_playing is not None and is_playing:
            logger.info(f"行情：{key} 在推送中...")
            return
        self.playing_data_dict[key] = True
        asyncio.create_task(self.play_data(param.market, param.code, param.date))
